package com.wyz.socket.core.impl;


import com.wyz.socket.core.api.IoProvider;
import com.wyz.socket.core.base.ProviderCallback;
import com.wyz.socket.core.utils.NameThreadFactory;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;


public class IoSelectorProvider implements IoProvider {
    private final AtomicBoolean isClosed = new AtomicBoolean(false); // 总开关

    // 是否处于某个过程
    private final Selector receiveSelector;
    private final AtomicBoolean inRegReceive = new AtomicBoolean(false);
    private final HashMap<SelectionKey, Runnable> receiveEventCallbackMap = new HashMap<>();
    private final ExecutorService receiveEventHandlePool;

    private final Selector sendSelector;
    private final AtomicBoolean inRegSend = new AtomicBoolean(false);
    private final HashMap<SelectionKey, Runnable> sendEventCallbackMap = new HashMap<>();
    private final ExecutorService sendEventHandlePool;


    public IoSelectorProvider() throws IOException { // 创建
        receiveSelector = Selector.open(); // 获取read的选择器
        sendSelector = Selector.open(); // 获取write 的选择器
        receiveEventHandlePool = Executors.newFixedThreadPool(4, new NameThreadFactory("IoProvider-Input-Thread-")); // 线程池
        sendEventHandlePool = Executors.newFixedThreadPool(4, new NameThreadFactory("IoProvider-Output-Thread-")); // 线程池
        // 开始输出输入的监听
        startRead();
        startWrite();
    }

    private void startRead() {
        Thread thread = new Thread("Clink IoSelectorProvider ReadSelector Thread") {
            @Override
            public void run() {
                while (!isClosed.get()) {
                    try {
                        if (receiveSelector.select() == 0) { // 判断是否有激活读取动作的事件
                            waitSelection(inRegReceive); // 延时处理
                            continue;
                        } else if (inRegReceive.get()) {
                            waitSelection(inRegReceive); // 延时处理
                        }

                        Iterator<SelectionKey> iterator = receiveSelector.selectedKeys().iterator();
                        while (iterator.hasNext()) {
                            SelectionKey selectionKey = iterator.next();
                            if (selectionKey.isValid()) { // 判断有效
                                handleSelection(selectionKey, SelectionKey.OP_READ, receiveEventCallbackMap, receiveEventHandlePool);
                            }
                            iterator.remove();
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }


        };
        thread.setPriority(Thread.MAX_PRIORITY);
        thread.start();
    }

    private void startWrite() {
        Thread thread = new Thread("Clink IoSelectorProvider WriteSelector Thread") {
            @Override
            public void run() {
                while (!isClosed.get()) {
                    try {
                        if (sendSelector.select() == 0) {
                            waitSelection(inRegSend);
                            continue;
                        } else if (inRegSend.get()) {
                            waitSelection(inRegSend);
                        }
                        Iterator<SelectionKey> iterator = sendSelector.selectedKeys().iterator();
                        while (iterator.hasNext()) {
                            SelectionKey selectionKey = iterator.next();
                            if (selectionKey.isValid()) {
                                handleSelection(selectionKey, SelectionKey.OP_WRITE, sendEventCallbackMap, sendEventHandlePool);
                            }
                            iterator.remove();
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        thread.setPriority(Thread.MAX_PRIORITY);
        thread.start();
    }

    @Override
    public void register(ProviderCallback callback) throws Exception {
        SelectionKey key = null;
        if (callback.ops == SelectionKey.OP_READ) {
            key = registerSelection( // 通过该注册方法 使得 channel 被包装为 selectorkey  callback 被包装为一个runnable
                    callback.channel,
                    receiveSelector, //并注册到readselect上
                    SelectionKey.OP_READ,
                    inRegReceive, // readselectot 注册锁 主要是将 readselector 从select()状态中释放,从而能够进行一系列注册操作
                    receiveEventCallbackMap,
                    callback);
        } else if (callback.ops == SelectionKey.OP_WRITE) {
            key = registerSelection(
                    callback.channel,
                    sendSelector,
                    SelectionKey.OP_WRITE,
                    inRegSend,
                    sendEventCallbackMap,
                    callback);
        }
        if (key == null) {
            throw new IOException("register error : channel = " + callback.channel + " ops = " + callback.ops);
        }
    }


    @Override
    public void unregister(SocketChannel channel) {
        unRegisterSelection(channel, receiveSelector, receiveEventCallbackMap, inRegReceive);
        unRegisterSelection(channel, sendSelector,    sendEventCallbackMap,    inRegSend);
    }


    @Override
    public void close() {
        if (isClosed.compareAndSet(false, true)) {
            receiveEventHandlePool.shutdown();
            sendEventHandlePool.shutdown();
            receiveEventCallbackMap.clear();
            sendEventCallbackMap.clear();
            receiveSelector.wakeup();
            sendSelector.wakeup();
            try {
                receiveSelector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                sendSelector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private static void waitSelection(final AtomicBoolean locker) {
        //noinspection SynchronizationOnLocalVariableOrMethodParameter
        synchronized (locker) {
            if (locker.get()) {
                try {
                    locker.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    private static SelectionKey registerSelection(SocketChannel channel, Selector selector,
                                                  int registerOps, AtomicBoolean locker,
                                                  HashMap<SelectionKey, Runnable> map,
                                                  Runnable runnable) {

        //noinspection SynchronizationOnLocalVariableOrMethodParameter
        synchronized (locker) {
            // 设置锁定状态
            locker.set(true);
            try {
                // 唤醒当前的selector，让selector不处于select()状态
                selector.wakeup();
                SelectionKey key = null;
                if (channel.isRegistered()) {
                    // 查询是否已经注册过
                    key = channel.keyFor(selector);
                    if (key != null) {
                        key.interestOps(key.interestOps() | registerOps);
                    }
                }
                if (key == null) {
                    // 注册selector得到Key
                    key = channel.register(selector, registerOps);
                    // 注册回调
                    map.put(key, runnable);
                }
                return key;
            } catch (ClosedChannelException e) {
                return null;
            } finally {
                // 解除锁定状态
                locker.set(false);
                try {
                    // 通知
                    locker.notify();
                } catch (Exception ignored) {
                }
            }
        }
    }

    private static void unRegisterSelection(SocketChannel channel, Selector selector,
                                            Map<SelectionKey, Runnable> map, final AtomicBoolean locker) {
        //noinspection SynchronizationOnLocalVariableOrMethodParameter
        synchronized (locker) {
            // 设置锁定状态
            locker.set(true);
            try {
                // 唤醒当前的selector，让selector不处于select()状态
                selector.wakeup();
                if (channel.isRegistered()) {
                    SelectionKey key = channel.keyFor(selector);
                    if (key != null) {
                        // 取消监听的方法
                        key.cancel();
                        map.remove(key);
                        selector.wakeup();
                    }
                }
            } finally {
                // 解除锁定状态
                locker.set(false);
                try {
                    // 通知
                    locker.notify();
                } catch (Exception ignored) {
                }
            }
        }
    }

    private static void handleSelection(SelectionKey key, int keyOps,
                                        HashMap<SelectionKey, Runnable> map,
                                        ExecutorService pool) {
        // 重点
        // 取消继续对keyOps的监听.
        key.interestOps(key.interestOps() & ~keyOps);
        Runnable runnable = null;
        try {
            runnable = map.get(key);
        } catch (Exception ignored) {

        }
        if (runnable != null && !pool.isShutdown()) {
            // 异步调度
            pool.execute(runnable);
        }
    }

}
